package com.gis.bigdata.spark02.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * @author LnnuUser
 * @create 2022-10-06-上午11:45
 */
object Scala_Test02_1 {

  def main(args: Array[String]): Unit = {
    // TOP10热门品类中每个品类的TOP10活跃session统计
    // 在需求1的基础上，增加每个品类用户session的点击统计

    // TODO: TOP10
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    val sc: SparkContext = new SparkContext(sparkConf)

    // TODO 1. 读取原始数据
    val actionRDD: RDD[String] = sc.textFile("/home/lnnu/IdeaProjects/spark/datas/user_visit_action.txt")
    actionRDD.cache()
    val top10Ids: Array[String] = top10Category(actionRDD)

    // 1. 过滤原始数据,保留点击和前10品类ID
    val filterAction: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        if (datas(6) != "-1") {
          top10Ids.contains(datas(6))
        } else {
          false
        }
      }
    )

    // 2. 根据品类ID和sessionId进行点击量的统计
    val reduceRDD: RDD[((String, String), Int)] = filterAction.map(
      action => {
        val datas: Array[String] = action.split("_")
        ((datas(6), datas(2)), 1)
      }
    ).reduceByKey(_ + _)

    // 3. 将统计的结果进行结构的转化
    // （（品类id， sessionid），sum） => （品类id，（sessionid， sum））
    val mapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
      case ((cid, sid), sum) => {
        (cid, (sid, sum))
      }
    }

    // 4. 相同的品类进行分组
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupByKey()

    // 5. 将分组后的数据进行点击量的排序，取前10
    val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      item => {
        item.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    )

    resultRDD.collect().foreach(println)

    sc.stop()
  }

  def top10Category(actionRDD: RDD[String]):  Array[String] = {

    val flatRDD: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap(
      action => {
        val datas: Array[String] = action.split("_")
        if (datas(6) != "-1") {
          // 点击
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          // 下单
          val ids: Array[String] = datas(8).split(",")
          ids.map(id => (id, (0, 1, 0)))
        } else if (datas(10) != "null") {
          // 支付
          val ids: Array[String] = datas(10).split(",")
          ids.map(id => (id, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )

    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey(
      (t1, t2) => {
        ((t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3))
      }
    )

    analysisRDD.sortBy(_._2, false).take(10).map(_._1)

  }

}
